package com.powernode.demo;

import com.powernode.constant.MqConstant;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.junit.Test;

import java.io.IOException;
import java.util.List;

/**
 * RocketMQ发送同步消息
 */
public class ASimpleTest {
    /**
     * 生产者
     */
    @Test
    public void simpleProducer() throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        // 创建生产者(指定一个组名)
        DefaultMQProducer producer = new DefaultMQProducer("test-producer-group");
        // 连接namesrv
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
        // 启动
        producer.start();
        // 创建一个消息
        Message message = new Message("testTopic", "我是一个简单的消息".getBytes());
        // 发送消息
        SendResult sendResult = producer.send(message);
        System.out.println("发送消息结果: " + sendResult.getSendStatus());
        // 关闭生产者
        producer.shutdown();
    }

    /**
     * 消费者
     * @throws MQClientException
     */
    @Test
    public void simpleConsumer() throws MQClientException, IOException {
        // 创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-producer-group");
        // 连接namesrv
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
        // 订阅主题(* 标识订阅这个主题中所有的消息)
        consumer.subscribe("testTopic", "*");
        // 设置监听器(异步回调方式)
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msg, ConsumeConcurrentlyContext context) {
                // 这是消费的方法(业务逻辑处理)
                System.out.println("我是消费者");
                System.out.println(msg.get(0).toString());
                System.out.println("消息内容" + new String(msg.get(0).getBody()));
                System.out.println("消费上下文: " + context);
                // 返回值  CONSUME_SUCCESS表示消费成功,消息会从mq出队; RECONSUME_LATER(报错/null)表示失败,消息会重新回到队列,过一会重新冲投递出来,会给当前消费者或其他消费者消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动
        consumer.start();
        // 挂起当前jvm
        System.in.read();
    }
}
